package cn.wzy.rocketmq.service.impl;

import cn.wzy.rocketmq.service.IRocketMqService;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;

@Service
public class RocketMqServiceImpl implements IRocketMqService {

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    @Value("${rocketmq.producer.send-message-timeout}")
    private Integer timeOut;

    /**
     * 发送带tag的消息,直接在topic后面加上":tag"
     * @param msg
     * @param topic
     * @param tag
     */
    @Override
    public void sendMsg(String msg,String topic,String tag) {
        //destination不一样其他都是一样的
        String destination = StringUtils.isNotBlank(tag )?topic+":"+tag:topic;
        rocketMQTemplate.syncSend(destination, MessageBuilder.withPayload(msg).build());
    }

    @Override
    public void sendMsg(String msg,String topic) {
        sendMsg(msg,topic,null);
    }


    @Override
    public void sendAsyncMsg(String msg,String topic,String tag) {
        //中高级程序员的写法
        String destination = StringUtils.isNotBlank(tag )?topic+":"+tag:topic;
        rocketMQTemplate.asyncSend(destination,
                MessageBuilder.withPayload(msg).build(),
                new RonghuanetSendCallback());

    }
    @Override
    public void sendAsyncMsg(String msg,String topic) {
        sendAsyncMsg(msg,topic,null);
    }

    @Override
    public void sendDelayMsg(String msg,String topic,String tag, int delayLevel) {
        // 初级程序员写法
        if (StringUtils.isNotBlank(tag ))
            rocketMQTemplate.syncSend(topic+":"+tag,
                    MessageBuilder.withPayload(msg).build(),
                    timeOut,
                    delayLevel
            );
        else
            rocketMQTemplate.syncSend(topic,
                    MessageBuilder.withPayload(msg).build(),
                    timeOut,
                    delayLevel);
    }

    @Override
    public void sendDelayMsg(String msg,String topic,int delayLevel) {
        sendDelayMsg(msg,topic,null,delayLevel);
    }


    /**
     * @param txProducerGroup 和listener的组一致
     * @param msg
     * @param topic
     * @param tag
     */
    @Override
    public void sendTransMsg(String txProducerGroup, String msg, String topic, String tag) {
        String destination = StringUtils.isNotBlank(tag ) ? topic + ":" + tag : topic;
        rocketMQTemplate.sendMessageInTransaction(txProducerGroup,destination,
                MessageBuilder.withPayload(msg).build(),null);
    }

    @Override
    public void sendTransMsg(String txProducerGroup, String msg, String topic) {
        sendTransMsg(txProducerGroup,msg,topic,null);
    }
}

class RonghuanetSendCallback implements SendCallback {

    @Override
    public void onSuccess(SendResult sendResult) {
        System.out.println("发送成功!!!!");
    }

    @Override
    public void onException(Throwable throwable) {
        System.out.println("发送失败!!!!");
    }
}